package com.minsf.rpc.proxy;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResource;
import cn.hutool.http.HttpResponse;
import com.minsf.rpc.RpcApplication;
import com.minsf.rpc.config.RpcConfig;
import com.minsf.rpc.constant.RpcConstant;
import com.minsf.rpc.fault.retry.RetryStrategy;
import com.minsf.rpc.fault.retry.RetryStrategyFactory;
import com.minsf.rpc.fault.tolerant.TolerantStrategy;
import com.minsf.rpc.fault.tolerant.TolerantStrategyFactory;
import com.minsf.rpc.loadbalancer.LoadBalancer;
import com.minsf.rpc.loadbalancer.LoadBalancerFactory;
import com.minsf.rpc.model.RpcRequest;
import com.minsf.rpc.model.RpcResponse;
import com.minsf.rpc.model.ServiceMetaInfo;
import com.minsf.rpc.protocol.*;
import com.minsf.rpc.registry.Registry;
import com.minsf.rpc.registry.RegistryFactory;
import com.minsf.rpc.serializer.JdkSerializer;
import com.minsf.rpc.serializer.Serializer;
import com.minsf.rpc.serializer.SerializerFactory;
import com.minsf.rpc.server.tcp.VertxTcpClient;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetClient;
import io.vertx.core.net.NetSocket;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;

/**
 * @Description 服务代理（ JDK 动态代理 ）
 * @Author minify
 * @Date 2024/3/24 17:18
 * @Project sf-rpc
 */
public class ServiceProxy implements InvocationHandler {

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        // 指定序列化器
        // Serializer serializer = new JdkSerializer();
        RpcConfig rpcConfig = RpcApplication.getRpcConfig();
        Serializer serializer = SerializerFactory.getSerializer(rpcConfig.getSerializer());

        // 构造请求
        RpcRequest rpcRequest = RpcRequest.builder()
                .args(args)
                .serviceName(method.getDeclaringClass().getName())
                .methodName(method.getName())
                .parameterTypes(method.getParameterTypes())
                .build();

        try {
            // 序列化
            byte[] bodyBytes = serializer.serialize(rpcRequest);

            // 从注册中心获取服务地址
            Registry registry = RegistryFactory.getInstance(rpcConfig.getRegistryConfig().getRegistry());
            ServiceMetaInfo serviceMetaInfo = new ServiceMetaInfo();
            serviceMetaInfo.setServiceName(rpcRequest.getServiceName());
            serviceMetaInfo.setServiceVersion(RpcConstant.DEFAULT_SERVICE_VERSION);

            List<ServiceMetaInfo> serviceMetaInfos = registry.serviceDiscovery(serviceMetaInfo.getServiceKey());
            if (CollectionUtil.isEmpty(serviceMetaInfos)) {
                throw new RuntimeException("暂无服务地址");
            }
            // 暂时获取第一个服务
            //ServiceMetaInfo serviceMetaInfo1 = serviceMetaInfos.get(0);

            // 发送请求
            // 注意这里被硬编码了（需要使用注册中心和服务发现机制解决）
            /*try (HttpResponse httpResponse = HttpRequest.post(serviceMetaInfo1.getServiceAddress())
                    .body(bodyBytes)
                    .execute()) {
                byte[] bytes = httpResponse.bodyBytes();
                // 反序列化
                RpcResponse rpcResponse = serializer.deserialize(bytes, RpcResponse.class);
                return rpcResponse.getData();
            }*/

            // 发送 TCP 请求
            /*Vertx vertx = Vertx.vertx();
            NetClient netClient = vertx.createNetClient();
            CompletableFuture<RpcResponse> responseFuture = new CompletableFuture<>();

            netClient.connect(serviceMetaInfo1.getServicePort(),
                    "localhost",
            result -> {
                if (result.succeeded()) {
                    System.out.println("Connected to TCP server");
                    NetSocket socket = result.result();

                    // 发送数据
                    // 构造消息
                    ProtocolMessage<RpcRequest> protocolMessage = new ProtocolMessage<>();
                    ProtocolMessage.Header header = new ProtocolMessage.Header();
                    header.setMagic(ProtocolConstant.PROTOCOL_MAGIC);
                    header.setVersion(ProtocolConstant.PROTOCOL_VERSION);
                    header.setSerializer((byte) ProtocolMessageSerializerEnum.getEnumByValue(RpcApplication.getRpcConfig().getSerializer()).getKey());
                    header.setType((byte) ProtocolMessageTypeEnum.REQUEST.getKey());
                    header.setRequestId(IdUtil.getSnowflakeNextId());
                    protocolMessage.setHeader(header);
                    protocolMessage.setData(rpcRequest);

                    // 编码请求
                    try {
                        Buffer encodeBuffer = ProtocolMessageEncoder.encode(protocolMessage);
                        socket.write(encodeBuffer);
                    } catch (Exception e) {
                        throw new RuntimeException("协议消息编码错误");
                    }

                    // 接收响应
                    socket.handler(buffer -> {
                        try {
                            ProtocolMessage<?> decoder = ProtocolMessageDecoder.decoder(buffer);
                            ProtocolMessage<RpcResponse> rpcResponseProtocolMessage = (ProtocolMessage<RpcResponse>) decoder;
                            // 如果任务没有完成，返回的值设置为给定值
                            responseFuture.complete(rpcResponseProtocolMessage.getData());
                        } catch (Exception e) {
                            throw new RuntimeException("协议消息解码错误");
                        }
                    });

                } else {
                    System.err.println("Failed to connect to TCP server");
                }
            });*/

            LoadBalancer loadBalancer = LoadBalancerFactory.getInstance(RpcApplication.getRpcConfig().getLoadBalancer());
            // 将方法名作为负载均衡器参数
            Map<String, Object> requestParams = new HashMap<>();
            requestParams.put("methodName", rpcRequest.getMethodName());
            ServiceMetaInfo selectServiceMetaInfo = loadBalancer.select(requestParams, serviceMetaInfos);

            RetryStrategy retryStrategy = RetryStrategyFactory.getInstance(RpcApplication.getRpcConfig().getRetryStrategy());
            RpcResponse rpcResponse = null;
            try {
                rpcResponse = retryStrategy.doRetry(() -> {
                    return VertxTcpClient.doRequest(rpcRequest, selectServiceMetaInfo); //responseFuture.get();
                });
            } catch (Exception e) {

                // 触发容错
                String tolerantStrategyKey = RpcApplication.getRpcConfig().getTolerantStrategy();
                TolerantStrategy tolerantStrategy = TolerantStrategyFactory.getInstance(tolerantStrategyKey);
                Map<String, Object> params = new HashMap<>();
                params.put("rpcRequest", rpcRequest);
                rpcResponse = tolerantStrategy.doTolerant(params, e);
            }

            // netClient.close();
            return rpcResponse.getData();

        } catch (Exception e) {
            e.printStackTrace();
        }

        return null;
    }
}
